package cn.kgc.app

import cn.kgc.bean.Ads_log
import cn.kgc.utils.{JdbcUtil, MyKafkaUtil}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.text.SimpleDateFormat
import java.util.Date

/**
 * 需求:实现实时的动态黑名单机制：将每天对某个广告点击超过 100 次的用户拉黑。
 *
 * 2）校验通过则对给用户点击广告次数累加一并存入MySQL；
 * 3）在存入MySQL之后对数据做校验，如果单日超过100次则将该用户加入黑名单。
 *
 */
object BlackListHandler {
  def main(args: Array[String]): Unit = {

    // 创建时间格式化转换对象  把时间转化为年月日
    val dateformat = new SimpleDateFormat("yyyy-MM-dd")

    // 创建sparkStreaming 环境
    val conf = new SparkConf().setMaster("local[*]").setAppName("BlackListHandler")
    val ssc = new StreamingContext(conf,Seconds(3))
    // 1）读取Kafka数据之后，并对MySQL中存储的黑名单数据做校验；
    val source = MyKafkaUtil.getKafkaStream("ad", ssc)
    val AdsLogDStream = source.map(x => {
      val arr = x.value().split(" ")

      Ads_log(arr(0).toLong, arr(1), arr(2), arr(3), arr(4))
    })

    // 去mysql 的黑名单表中查询是否有此用户id  如果没有就统计到广告点击统计表总

    AdsLogDStream.foreachRDD(x =>{
      x.foreachPartition(it =>{
        // 获取连接对象
        val conn = JdbcUtil.getConnection
        it.foreach(rdd =>{
          val sql = "select * from  black_list where userid = ?"
          val flag = JdbcUtil.isExist(conn, sql, Array(rdd.userid))
          // 如果没有数据就在user_ad_count中添加数据
          if(!flag) {
            // 查询user_ad_count中是否有相同的用户和广告和时间  没有的话直接插入 有 更新
            /*val bool = JdbcUtil.isExist(
              conn,
              "select * from user_ad_count where dt = ? and userid = ? and adid = ?",
              Array(dateformat.format(new Date(rdd.timestamp)),rdd.userid, rdd.adid))
            if(bool){
              JdbcUtil.executeUpdate(
                conn,
                "update user_ad_count set count = count + 1   where userid = ? and adid = ? ",
                Array(rdd.userid, rdd.adid)
              )
            }else{

              JdbcUtil.executeUpdate(
                conn,
                "insert into user_ad_count values(?,?,?,?)",
                Array(dateformat.format(new Date(rdd.timestamp)),rdd.userid,rdd.adid,1)
              )
            }*/
            JdbcUtil.executeUpdate(
              conn,
              """
                | insert into user_ad_count(dt,userid,adid,count) values(?,?,?,?)
                | ON DUPLICATE KEY
                | update count = count + 1
                |""".stripMargin,
              Array(dateformat.format(new Date(rdd.timestamp)), rdd.userid, rdd.adid, 1)
            )


            // 添加数据之后做黑名单的更新
            // 如果那个用户在一天之内点击广告次数超过100 就将此用户加入到黑名单中

            val count = JdbcUtil.getDataFromMysql(
              conn,
              "select count from user_ad_count where userid = ? and adid = ?",
              Array(rdd.userid, rdd.adid)
            )
            if(count >= 100){
              // 对黑名单表进行添加
              JdbcUtil.executeUpdate(conn,"insert into black_list values(?)",Array(rdd.userid))
            }
          }
        })
        conn.close()
      })
    })



    ssc.start()

    ssc.awaitTermination()

  }


}
